# -*- coding:utf-8 -*-
from __future__ import absolute_import, unicode_literals

import arrow
import requests
from celery import Celery, platforms
from raven import Client
from raven.contrib.celery import register_logger_signal, register_signal

import conf
from connection import InitConnectionTask, pg_insert, pg_select, pg_update

client = Client(conf.SENTRY_DNS)
register_logger_signal(client)
register_signal(client, ignore_expected=True)

platforms.C_FORCE_ROOT = True

sync_flow = Celery('sync_flow_worker', broker=conf.MQ_BROKER)

sync_flow.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Asia/Shanghai',
    worker_max_tasks_per_child=200,
    worker_disable_rate_limits=True,
    task_create_missing_queues=True,
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    broker_transport_options={'confirm_publish': True}
)


class SyncFlowTask(InitConnectionTask):
    abstract = True
    threshold_timestamp = None
    map_key = {
        'appCode': 'app_code',
        'chargeId': 'charge_id',
        'createdAt': 'created_at',
        'id': 'order_id',
        'orderType': 'order_type',
        'platformCode': 'platform_code',
        'refundSource': 'refund_source',
        'refundedAt': 'refunded_at',
        'serialNum': 'flow_id',
        'userId': 'user_id',
        'userPhone': 'user_phone',
        'additionalPrice': 'additional_price',
        'orderPrice': 'order_price',
        'paidAt': 'paid_at',
        'payPrice': 'pay_price'
    }

    @property
    def prdb_0(self):
        return super().prdb(0)

    @property
    def prdb_1(self):
        return super().prdb(1)

    @property
    def prdb_2(self):
        return super().prdb(2)

    @property
    def pgdb(self):
        return super().pgdb('account')

@sync_flow.task(base=SyncFlowTask, bind=True, ignore_result=True, max_retries=3)
def ping_worker(self, key=None):
    """ping-pong
        key (str, optional): Defaults to None. redis.key

    Returns:
        list: key,pong
    """

    if key is None:
        key = 'monitor:' + self.name.replace('.', ':')
    self.rdb.set(key, 'Pong', ex=600)

    return key, 'Pong!'

@sync_flow.task(base=SyncFlowTask, bind=True, ignore_result=True, max_retries=3)
def pg_upsert_by_flow_id(self, flow_key, task_name, app_code_order_id=None):
    """同步单个flow_key
    Args:
        flow_key(str): p:91ala:ala212994754455289
        task_name(int): 1 or 2
        app_code_order_id(str): 91ala:ala004254793203734
    """
    # 1.获取数据
    if task_name == 1:
        data = self.prdb_1.hgetall(flow_key)
    elif task_name == 2:
        if app_code_order_id:
            for flow_id in self.prdb_0.smembers('o_r_mapper:{}'.format(app_code_order_id)):
                flow_key = 'r:{}:{}'.format(app_code_order_id.split(':')[0], flow_id)
                pg_upsert_by_flow_id.apply_async(args=[flow_key, task_name], queue=conf.MQ_SYNC_FLOW)
            return app_code_order_id, 'fetch flowKeySet success'
        data = self.prdb_2.hgetall(flow_key)
    else:
        data = None
    if not data:
        return flow_key, 'is null'
    if 'serialNum' not in data:
        data['serialNum'] = flow_key.split(':')[-1]
    # 2.拼装数据
    flow_id = data['serialNum']
    table_name = 'pay' if task_name == 1 else 'refund'
    select_data, keys = pg_select("SELECT * FROM {} WHERE flow_id='{}';".format(table_name, flow_id), self.pgdb, field=True)
    map_key = {
        'appCode': 'app_code',
        'chargeId': 'charge_id',
        'createdAt': 'created_at',
        'id': 'order_id',
        'orderType': 'order_type',
        'platformCode': 'platform_code',
        'refundSource': 'refund_source',
        'refundedAt': 'refunded_at',
        'serialNum': 'flow_id',
        'userId': 'user_id',
        'userPhone': 'user_phone',
        'additionalPrice': 'additional_price',
        'orderPrice': 'order_price',
        'paidAt': 'paid_at',
        'payPrice': 'pay_price'
    }
    flow_data = {map_key.get(k, k): v for k, v in data.items() if map_key.get(k, k) in keys and v}
    flow_keys = list(flow_data.keys())
    # 3.同步数据
    if select_data:
        sql_set = '=%s,'.join(flow_keys) + '=%s'
        sql = "UPDATE {} SET {} WHERE flow_id=%s;".format(table_name, sql_set)
        pg_update(sql, [flow_data], self.pgdb)
        return flow_key, 'update success'
    else:
        no_null_keys = set(['flow_id', 'order_id', 'user_id', 'app_code', 'created_at', 'price', 'order_type', 'status'])
        if no_null_keys.issubset(set(flow_keys)):
            sql_set = ','.join(keys)
            s = ','.join(['%s'] * len(keys))
            sql = "INSERT INTO {} ({}) VALUES ({});".format(table_name, sql_set, s)
            pg_insert(sql, [flow_data], self.pgdb)
            return flow_key, 'insert success'
        else:
            redis_flow_compensate.apply_async(args=[flow_key, task_name], queue=conf.MQ_SYNC_FLOW)
            raise self.retry(countdown=10, exc=KeyError(flow_key, 'missing fields, call compensate'))
            return flow_key, 'missing fields, UPSET failure'


@sync_flow.task(base=SyncFlowTask, bind=True, ignore_result=True, max_retries=3)
def redis_flow_compensate(self, flow_key, task_name):
    """8个非空字段的补偿。 Warning!Warning!Warning! 异常的流水信息会被清理。
        flow_key (str, optional): Defaults to None. p:*:*
        task_name (int, optional): Defaults to None. 1:payment.1, 2:payment.2

    Returns:
        list: -
    """

    if flow_key is None:
        return None
    rdb = self.prdb_1 if task_name == 1 else self.prdb_2
    flow_message = rdb.hgetall(flow_key)

    if not flow_message:
        return flow_key, task_name, 'redis.flow is null'

    if not flow_message.get('id'):
        rdb.delete(flow_key)
        return flow_key, task_name, 'redis.flow.id is null and delete success'

    lite_list_data = self.mdb.order_lite_list.find_one({'id': flow_message['id']},
                                                       {'orderType': 1,
                                                        'createdAt': 1,
                                                        'appCode': 1,
                                                        'userId': 1,
                                                        '_id': 0})
    if not lite_list_data:
        return flow_key, task_name, 'mongo.lite_list is None'
    # 因为redis存贮都是以字符串的形式，如'0'，所以这里就简单的以get()来判断
    if not flow_message.get('orderType'):
        rdb.hset(flow_key, 'orderType', lite_list_data['orderType'])

    if not flow_message.get('createdAt'):
        rdb.hset(flow_key, 'createdAt', int(lite_list_data['createdAt']))

    if not flow_message.get('appCode'):
        rdb.hset(flow_key, 'appCode', lite_list_data.get('appCode', 'boluome'))

    if not flow_message.get('userId'):
        rdb.hset(flow_key, 'userId', lite_list_data['userId'])

    order_data = self.mdb['order_{}'.format(lite_list_data['orderType'])].find_one({'id': flow_message['id']},
                                                                                   {'orderPrice': 1,
                                                                                    'price': 1,
                                                                                    'paidList': 1,
                                                                                    '_id': 0})
    if not order_data:
        return flow_key, task_name, 'mongo.order is None'

    flow_id = flow_key.split(':')[-1]

    if not flow_message.get('serialNum'):
        rdb.hset(flow_key, 'serialNum', flow_id)

    if not flow_message.get('status'):
        if flow_id in order_data.get('paidList', []):
            rdb.hset(flow_key, 'status', 2)
        else:
            rdb.hset(flow_key, 'status', 0)

    if not flow_message.get('orderPrice'):
        rdb.hset(flow_key, 'orderPrice', int(order_data['orderPrice'] * 100))

    if not flow_message.get('price'):
        rdb.hset(flow_key, 'price', int(order_data['price'] * 100))

    return flow_key, 'compensate success'


@sync_flow.task(base=SyncFlowTask, bind=True, ignore_result=True, max_retries=3)
def redis_flow_pay_check_up(self, app_code=None, flow_key=None):
    """每月一号定时遍历redis.flow支付和退款，检查未同步至pg的流水，并补推
        app_code (str, optional): Defaults to None. "bestpay"
        flow_key (str, optional): Defaults to None. "p:*:*"
    """
    if app_code is None:
        now = arrow.now()
        old = now.shift(months=-3)
        SyncFlowTask.threshold_timestamp = old.timestamp * 1000
        for app_code in self.mdb.order_lite_list.distinct('appCode'):
            if not app_code:
                continue
            redis_flow_pay_check_up.apply_async(args=[app_code], queue=conf.MQ_SYNC_FLOW)
            redis_flow_refund_check_up.apply_async(args=[app_code], queue=conf.MQ_SYNC_FLOW)
        return 'appCode distribution completed'
    if flow_key is None:
        for flow_key in self.prdb_1.scan_iter("p:{}:*".format(app_code)):
            redis_flow_pay_check_up.apply_async(args=[app_code, flow_key], queue=conf.MQ_SYNC_FLOW)
        return app_code, 'flowKey distribution completed'

    flow_message = self.prdb_1.hgetall(flow_key)
    if not flow_message:
        return flow_key, 'flowMessage is {}'

    if 'createdAt' not in flow_message or 'status' not in flow_message or 'serialNum' not in flow_message:
        # 基本信息都没有，让下面这个函数去做补偿或删除
        redis_flow_compensate.apply_async(args=[flow_key, 1], queue=conf.MQ_SYNC_FLOW)
        raise self.retry(countdown=10, exc=KeyError(flow_key, 'flowMessage missing field and compensate'))
        return flow_key, 'flowMessage missing field'

    # 先看看pg.boluome.pay 有没有准确的存储这笔流水信息。
    pg_data = list(pg_select("SELECT * FROM pay WHERE flow_id='{}';".format(flow_message['serialNum']), self.pgdb))
    if not pg_data:
        # 没有那就同步
        pg_upsert_by_flow_id.apply_async(args=[flow_key, 1], queue=conf.MQ_SYNC_FLOW)
        raise self.retry(countdown=10, exc=KeyError(flow_key, 'pg.pay not data and sync'))
    else:
        # 有，但是数据不匹配，那就同步
        flow_data = {self.map_key.get(k, k): v for k, v in flow_message.items()}
        for k, v in pg_data[0].items():
            if not flow_data.get(k):
                continue
            elif v == flow_data[k]:
                continue
            elif isinstance(v, int) and v == int(flow_data[k]):
                continue
            elif isinstance(v, float) and v == float(flow_data[k]):
                continue
            pg_upsert_by_flow_id.apply_async(args=[flow_key, 1], queue=conf.MQ_SYNC_FLOW)
            raise self.retry(countdown=10, exc=KeyError(flow_key, k, v, flow_data.get(k), 'sync pg.pay'))

    if self.threshold_timestamp is None:
        now = arrow.now()
        old = now.shift(months=-3)
        SyncFlowTask.threshold_timestamp = old.timestamp * 1000

    # 下面关注清结算系统
    if int(flow_message['createdAt']) > self.threshold_timestamp:
        return flow_key, 'within three months, not synchronized'

    if flow_message['status'] == '0':
        # delete redis.flow
        self.prdb_1.delete(flow_key)
        return flow_key, 'delete completed'

    flow_id = flow_message['serialNum']
    if list(pg_select(f"SELECT flow_id FROM boluome_flow WHERE flow_id='{flow_id}';", self.pgdb)):
        # delete redis.flow
        self.prdb_1.delete(flow_key)
        return flow_key, 'delete completed'

    body = {
        'exchangeName': 'flow',
        'routingKey': 'payment',
        'payload': flow_message
    }
    requests.post(conf.RABBITMQAPI + '/rabbitmq/publish', json=body)
    raise self.retry(countdown=10, exc=KeyError(flow_key, 'sync pg.settlement'))
    return flow_key, 'sync settlement success'


@sync_flow.task(base=SyncFlowTask, bind=True, ignore_result=True, max_retries=3)
def redis_flow_refund_check_up(self, app_code, flow_key=None):
    """每月一号定时遍历redis.flow支付和退款，检查未同步至pg的流水，并补推
        app_code (str, optional): Defaults to None. "bestpay"
        flow_key (str, optional): Defaults to None. "p:*:*"
    """
    if flow_key is None:
        for flow_key in self.prdb_2.scan_iter("r:{}:*".format(app_code)):
            redis_flow_refund_check_up.apply_async(args=[app_code, flow_key], queue=conf.MQ_SYNC_FLOW)
        return app_code, 'flowKey distribution completed'

    flow_message = self.prdb_2.hgetall(flow_key)
    if not flow_message:
        return flow_key, 'flowMessage is {}'

    if 'createdAt' not in flow_message or 'status' not in flow_message or 'serialNum' not in flow_message:
        # 基本信息都没有，让下面这个函数去做补偿或删除
        redis_flow_compensate.apply_async(args=[flow_key, 2], queue=conf.MQ_SYNC_FLOW)
        raise self.retry(countdown=10, exc=KeyError(flow_key, 'flowMessage missing field and compensate'))
        return flow_key, 'flowMessage missing field'

    # 先看看pg.boluome.pay 有没有准确的存储这笔流水信息。
    pg_data = list(pg_select("SELECT * FROM refund WHERE flow_id='{}';".format(flow_message['serialNum']), self.pgdb))
    if not pg_data:
        # 没有那就同步
        pg_upsert_by_flow_id.apply_async(args=[flow_key, 2], queue=conf.MQ_SYNC_FLOW)
        raise self.retry(countdown=10, exc=KeyError(flow_key, 'pg.refund not data and sync'))
    else:
        # 有，但是数据不匹配，那就同步
        flow_data = {self.map_key.get(k, k): v for k, v in flow_message.items()}
        for k, v in pg_data[0].items():
            if not flow_data.get(k):
                continue
            elif v == flow_data[k]:
                continue
            elif isinstance(v, int) and v == int(flow_data[k]):
                continue
            elif isinstance(v, float) and v == float(flow_data[k]):
                continue
            pg_upsert_by_flow_id.apply_async(args=[flow_key, 2], queue=conf.MQ_SYNC_FLOW)
            raise self.retry(countdown=10, exc=KeyError(flow_key, k, v, flow_data.get(k), 'sync pg.refund'))

    if self.threshold_timestamp is None:
        now = arrow.now()
        old = now.shift(months=-3)
        SyncFlowTask.threshold_timestamp = old.timestamp * 1000

    # 下面关注清结算系统
    if int(flow_message['createdAt']) > self.threshold_timestamp:
        return flow_key, 'within three months, not synchronized'

    if flow_message['status'] == '0':
        # delete redis.flow
        self.prdb_2.delete(flow_key)
        return flow_key, 'delete completed'

    flow_id = flow_message['serialNum']
    if list(pg_select(f"SELECT flow_id FROM boluome_refund_flow WHERE flow_id='{flow_id}';", self.pgdb)):
        # delete redis.flow
        self.prdb_2.delete(flow_key)
        return flow_key, 'delete completed'

    body = {
        'exchangeName': 'flow',
        'routingKey': 'refund',
        'payload': flow_message
    }
    requests.post(conf.RABBITMQAPI + '/rabbitmq/publish', json=body)
    raise self.retry(countdown=10, exc=KeyError(flow_key, 'sync pg.settlement'))
    return flow_key, 'sync settlement success'